package xyz.xuw.storm.workcount.bolt;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import xyz.xuw.storm.workcount.model.MyMessage;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by xuw on 2016/11/10.
 */
public class WordCounterBolt extends BaseBasicBolt {
    Logger LOGGER = LoggerFactory.getLogger(getClass());
    private Integer id;
    private String name;
    Map<String, Integer> counters;

    @Override
    public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {
        MyMessage message = (MyMessage) tuple.getValues().get(0);
        String str = message.getWord();
        if (!counters.containsKey(str)) {
            counters.put(str, 1);
        } else {
            Integer c = counters.get(str) + 1;
            counters.put(str, c);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    @Override
    public void cleanup() {
        LOGGER.info("==word counter:name:{},id:{}==", name, id);
        for (Map.Entry<String, Integer> entry : counters.entrySet()) {
            LOGGER.info("key:{},value:{}", entry.getKey(), entry.getValue());
        }
    }

    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        this.counters = new HashMap<String, Integer>();
        this.name = context.getThisComponentId();
        this.id = context.getThisTaskId();
    }
}
